package com.c6906.vdam.dcs.service;

import com.c6906.vdam.dcs.pool.ProducerPool;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Service
public class ProducerService {
    private static final Logger log = LoggerFactory.getLogger(ProducerService.class);

    @Value("${mq.endpoint}")
    private String endpoint;

    @Value("${mq.producerTopic}")
    private String topic;

    private ProducerPool producerPool;

    @PostConstruct
    public void init() throws ClientException {
        producerPool = new ProducerPool(endpoint, topic);
    }

    @PreDestroy
    public void destroy() {
        producerPool.close();
    }

    public void sendMessageAsync(String topic, String tag, String key, byte[] body) throws Exception {
        Producer producer = producerPool.borrowProducer();
        try {
            ClientServiceProvider provider = ClientServiceProvider.loadService();
            Message message = provider.newMessageBuilder()
                    .setTopic(topic)
                    .setTag(tag)
                    .setKeys(key)
                    .setBody(body)
                    .build();
            producer.sendAsync(message);
        } finally {
            producerPool.returnProducer(producer);
        }
    }

    public void sendMessageSync(String topic, String tag, String key, byte[] body) throws Exception {
        Producer producer = producerPool.borrowProducer();
        try {
            ClientServiceProvider provider = ClientServiceProvider.loadService();
            Message message = provider.newMessageBuilder()
                    .setTopic(topic)
                    .setTag(tag)
                    .setKeys(key)
                    .setBody(body)
                    .build();
            producer.send(message);
        } finally {
            producerPool.returnProducer(producer);
        }
    }
}
